//
// Created by mio on 25-4-29.
//

#include "SigConnection.h"
#include "ConnectionManager.h"

SigConnection::SigConnection(TaskScheduler* scheduler, int socket)
    : TcpConnection(scheduler, socket),
        state_(NONE)
{
    this->SetReadCallback([this](std::shared_ptr<TcpConnection> conn, BufferReader& buffer)
    {
        return this->OnRead(buffer);
    });

    this->SetCloseCallback([this](std::shared_ptr<TcpConnection> conn)
    {
        this->DisConnected();
    });
}

SigConnection::~SigConnection()
{
    Clear();
}

void SigConnection::DisConnected()
{
    Clear();
}

// 添加一个关联的客户端标识符 (对方的 code_) 到 objectes_ 列表
// 表示当前连接与 code 标识的客户端建立了某种交互关系（如A拉B的流）
void SigConnection::AddCustom(const std::string& code)
{
    // 添加客户, 可能是拉流也可能时推流
    for (const auto& idefy : objectes_)
    {
        if (idefy == code) return; // 目标客户端已经添加
    }
    objectes_.push_back(code);
}

void SigConnection::RmoveCustom(const std::string& code)
{
    if (objectes_.empty()) return;
    objectes_.erase(std::remove(objectes_.begin(), objectes_.end(), code), objectes_.end());
    if (objectes_.empty())
    {
        state_ = IDLE; // 说明当前没有控制端, 自己也不是控制端
    }
}

bool SigConnection::OnRead(BufferReader& buffer)
{
    while (buffer.ReadableBytes() > 0) HandleMessage(buffer);
    return true;
}

void SigConnection::HandleMessage(BufferReader& buffer)
{
    if (buffer.ReadableBytes() < sizeof(packet_head)) return; // 数据不完整
    packet_head* data = (packet_head*)buffer.Peek();
    if (buffer.ReadableBytes() < data->len) return; // 数据还是不完整
    switch (data->cmd)
    {
    case JOIN:
        HandleJoin(data);           // 处理加入房间/身份验证请求
        break;
    case OBTAINSTREAM:
        HandleObtainStream(data);   // 处理请求获取/拉取流的请求
        break;
    case CREATESTREAM:
        HandleCreateStream(data);   // 此命令通常由被要求推流的客户端在成功创建流后，携带流地址发送给服务器
        break;
    case DELETESTREAM:
        HandleDeleteStream(data);   // 处理删除/停止流的请求
        break;
    case MOUSE:                     // 鼠标点击事件
    case MOUSEMOVE:                 // 鼠标移动事件
    case KEY:                       // 键盘事件
    case WHEEL:                     // 鼠标滚轮事件
        HandleOtherMessage(data);
        break;
    default:
        break;
    }
    // 更新缓冲区buffer
    buffer.Retrieve(data->len);
}

void SigConnection::Clear()
{
    state_ = CLOSE;
    conn_ = nullptr;
    DeleteStream_body body; // 通知所有关心者该删除流通知
    for (auto iter : objectes_)
    {
        // 从 ConnectionManager 中查询关联的对方客户端连接。
        TcpConnectionPtr conn = ConnectionManager::GetInstance()->QueryConn(iter);
        if (conn)   // 如果对方连接仍然存在
        {
            auto ctrConn = std::dynamic_pointer_cast<SigConnection>(conn);
            //    通知对方，我已经 (code_) 断开了，
            //    让对方从其 objectes_ 列表中移除我 (code_)。
            if (ctrConn) ctrConn->RmoveCustom(code_);
            //    设置通知消息中的 streamCount 为对方在其 RmoveCustom 之后，剩余的关联者数量。
            //    这个 streamCount 的语义是：对方现在还有多少个与之交互的连接。
            //    如果为0，且对方是PUSHER，它会停止推流。
            body.SetStreamCount(ctrConn->objectes_.size());
            conn->Send((const char*)&body, body.len);
        }
    }

    objectes_.clear();
    ConnectionManager::GetInstance()->RmvConn(code_);
    printf("Conn size : %d\n", ConnectionManager::GetInstance()->Size());
}

void SigConnection::HandleJoin(const packet_head* data)
{
    // 准备一个应答
    JoinReply_body reply_body;
    Join_body* body = (Join_body*)data;
    if (this->IsNoJoin()) // 当前连接是否处于 NONE 状态 (即尚未加入或验证过)。
    {
        std::string code = body->GetId(); // 获取客户端请求中携带的 ID。
        TcpConnectionPtr ptr = ConnectionManager::GetInstance()->QueryConn(code);
        if (ptr)
        {
            reply_body.SetCode(ERROR);
            this->Send((const char*)&reply_body, reply_body.len);
            return;
        }
        code_ = code;   // 设置当前 SigConnection 实例的唯一标识符 code_
        state_ = IDLE;  // 将状态设置为 IDLE (空闲，已验证，等待后续操作)。
        // 将当前连接添加到 ConnectionManager 中。
        ConnectionManager::GetInstance()->AddConn(code, shared_from_this());
        printf("Join Count : %d\n", ConnectionManager::GetInstance()->Size());
        reply_body.SetCode(SUCCESSFUL); // 设置应答码为成功
        this->Send((const char*)&reply_body, reply_body.len); // 发送成功应答
        return;
    }
    // 如果当前连接状态不是 NONE (例如已经 JOIN 过了，或者处于其他状态),则认为是非法操作或重复 JOIN，返回错误。
    reply_body.SetCode(ERROR);
    this->Send((const char*)&reply_body, reply_body.len);
}

// 处理客户端的 OBTAINSTREAM (请求获取/拉取其他客户端的流) 请求
void SigConnection::HandleObtainStream(const packet_head* data)
{
    ObtainStream_body* body = (ObtainStream_body*)data;
    this->DoObtainStream(body);
}

// 处理客户端的 CREATESTREAM 请求
// 这个消息通常是由一个被要求推流的客户端（我们称之为“推流端”）
// 在它成功与媒体服务器建立推流后，将媒体服务器提供的流地址
// 通过一个 CREATESTREAM 命令（消息体结构是 CreateStreamReply_body，因为它携带了地址）
// 发送给信令服务器，信令服务器的此 SigConnection 实例（即推流端自身）收到此消息。
void SigConnection::HandleCreateStream(const packet_head* data)
{
    CreateStream_body* body = (CreateStream_body*)data;
    this->DoCreateStream(body);
}

void SigConnection::HandleDeleteStream(const packet_head* data)
{
    if (this->IsBusy())
    {
        Clear();
    }
}

void SigConnection::HandleOtherMessage(const packet_head* data)
{
    // 鼠标、键盘消息、转发

    // 检查条件：
    // 1. conn_ 是否有效：conn_ 指向的是 PUSHER (被控端) 的连接。如果为 nullptr，说明当前不是 PULLER 或者 PULLER 未成功连接到 PUSHER。
    // 2. state_ 是否为 PULLER：只有 PULLER (控制端) 才有资格发送控制指令给 PUSHER。
    if (conn_ && state_ == PULLER) // 说明当前是在拉流
    {
        conn_->Send((const char*)data, data->len);
    }
}

// 调用者是客户端A (this)，它想拉取客户端B (target_client_code) 的流
void SigConnection::DoObtainStream(const packet_head* data)
{
    ObtainStreamReply_body reply;       // 准备给请求者A的应答消息。
    CreateStream_body create_reply;      // CreateStream_body 是一个不包含流地址的、请求对方创建流的信令
    std::string code = ((ObtainStream_body*)data)->GetId();  // 从收到的 ObtainStream_body 中获取目标客户端B的 ID
    TcpConnectionPtr conn = ConnectionManager::GetInstance()->QueryConn(code);
    if (!conn)
    {
        printf("Remote User don't exist\n");
        reply.SetCode(ERROR);
        this->Send((const char*)&reply, reply.len);
        return;
    }
    if (conn == shared_from_this()) // 不能控制自己
    {
        reply.SetCode(ERROR);
        this->Send((const char*)&reply, reply.len);
        return;
    }

    if (this->IsIdle()) // 检查当前客户端A (请求者) 是否处于 IDLE 状态
    {
        auto sigconn = std::dynamic_pointer_cast<SigConnection>(conn);
        // 根据目标客户端B 的状态进行处理
        switch (sigconn->GetRoleState())
        {
        case IDLE:      // 目标B也处于 IDLE 状态 (可以被控制，可以开始推流)
            this->state_ = PULLER;      // A的状态变为 PULLER (拉流端/控制端)
            this->AddCustom(code);      // A的 objectes_ 列表添加B的 ID
            sigconn->AddCustom(code_);  // B的 objectes_ 列表添加A的 ID (code_ 是A的ID)。
            reply.SetCode(SUCCESSFUL);
            conn_ = conn;               // A的 conn_ 指针指向B的连接，用于后续转发
            //    向目标客户端B (target_sig_conn) 发送 CREATESTREAM 请求，要求B去创建流。
            //    B收到这个请求后：
            //    - 与媒体服务器交互，创建RTMP等推流。
            //    - 成功后，向信令服务器发回一个携带流地址的 CREATESTREAM 消息 (实际上是 CreateStreamReply_body 结构的数据)。
            //    - 信令服务器收到B的这个消息后，会调用B的 SigConnection 实例的 HandleCreateStream -> DoCreateStream 方法。
            sigconn->Send((const char*)&create_reply, create_reply.len);
            this->Send((const char*)&reply, reply.len);                     // 向请求者A发送 OBTAINSTREAM 成功的应答
            break;
        case NONE:
        case CLOSE:
            printf("Target User is offline\n");
            reply.SetCode(ERROR);
            this->Send((const char*)&reply, reply.len);
            break;
        case PULLER:    // 目标B已经是 PULLER (即它正在控制别人/拉别人的流)
            printf("Target User is busy\n");
            reply.SetCode(ERROR);
            this->Send((const char*)&reply, reply.len);
            break;
        case PUSHER:    // 目标B已经是 PUSHER (即它正在推流)
            if (sigconn->GetStreamAddres().empty())
            {
                printf("Address is ERROR\n");
                reply.SetCode(ERROR);
                this->Send((const char*)&reply, reply.len);
            }
            else
            {
                // 目标B正在推流，并且有流地址
                // 更新请求者A的状态和关系：
                this->state_ = PULLER;              // A的状态变为 PULLER
                this->AddCustom(code);              // A的 objectes_ 添加B的 ID
                sigconn->AddCustom(code_);          // B (PUSHER) 的 objectes_ 添加新的拉流者A的 ID。
                // 构造一个 PLAYSTREAM 消息，包含B已存在的流地址，直接发送给请求者A。
                // A收到后就可以用这个地址去播放RTMP流了。
                PlayStream_body play_body;
                play_body.SetCode(SUCCESSFUL);
                play_body.SetStreamAddres(sigconn->GetStreamAddres());
                this->Send((const char*)&play_body, play_body.len);
            }
            break;
        default:
            break;
        }
    }
    else
    {
        reply.SetCode(ERROR);
        this->Send((const char*)&reply, reply.len);
    }
}

// 执行创建流的核心逻辑 (实际上是处理“推流端”上报其已创建的流信息)
// 当一个客户端 (我们称之为“推流端X”，即这里的 `this` SigConnection 实例)
// 被要求推流 (例如，收到了由 DoObtainStream 发出的 request_target_to_create_stream)，
// 并且它成功在媒体服务器上创建了推流后，它会向信令服务器发送一个 CREATESTREAM 命令。
// 这个命令的消息体结构是 CreateStreamReply_body，因为它携带了推流端X生成的流地址。
// 此函数 (DoCreateStream) 就是在推流端X的 SigConnection 实例上处理这个消息的。
void SigConnection::DoCreateStream(const packet_head* data)
{
    PlayStream_body body; // 准备一个 PlayStream_body 消息，用于通知所有等待此流的 PULLER 客户端去播放
    CreateStreamReply_body* reply = (CreateStreamReply_body*) data;
    streamAddres_ = reply->GetStreamAddres();
    //    遍历当前推流端X (this) 的 objectes_ 列表。
    //    这个列表包含了所有之前请求拉取X的流的 PULLER 客户端的 ID。
    //    需要通知它们流已创建 (或创建失败)
    for (auto& idefy : objectes_)
    {
        TcpConnectionPtr conn = ConnectionManager::GetInstance()->QueryConn(idefy);
        if (!conn)
        {
            this->RmoveCustom(idefy);
            continue;
        }
        auto sigconn = std::dynamic_pointer_cast<SigConnection>(conn);
        if (streamAddres_.empty()) // 流地址异常
        {
            printf("stream Address ERROR\n");
            sigconn->state_ = IDLE;
            body.SetCode(ERROR);
            sigconn->Send((const char*)&body, body.len);
        }
        switch (sigconn->GetRoleState())
        {
        case NONE:
        case IDLE:
        case CLOSE:
        case PUSHER:
            body.SetCode(ERROR);
            this->RmoveCustom(sigconn->GetCode());
            sigconn->Send((const char*)&body, body.len);
            break;
        case PULLER:
            this->state_ = PULLER;
            body.SetCode(SUCCESSFUL);
            body.SetStreamAddres(streamAddres_);
            printf("streamadder : %s\n", streamAddres_.c_str());
            sigconn->Send((const char*)&body, body.len);
            break;
        default:
            break;
        }
    }
}
